package com.peng.rpc.serializer;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.peng.rpc.entity.RpcRequest;
import com.peng.rpc.entity.RpcResponse;
import com.peng.rpc.enumeration.RpcError;
import com.peng.rpc.exception.RpcException;
import lombok.extern.slf4j.Slf4j;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;

/**
 * @author Wonderson
 * @email wpp.ds@foxmail.com
 * @date 2022/3/2 21:17
 */
@Slf4j
public class KryoSerializer implements CommonSerializer{

    private final ThreadLocal<Kryo> kryoThreadLocal = ThreadLocal.withInitial(() -> {
        Kryo kryo = new Kryo();
        kryo.register(RpcRequest.class);
        kryo.register(RpcResponse.class);
        kryo.setReferences(true);
        kryo.setRegistrationRequired(false);
        return kryo;
    });

    @Override
    public byte[] serialize(Object object) {
        try (
                ByteArrayOutputStream baos = new ByteArrayOutputStream();
                Output output = new Output(baos);
        ) {
            Kryo kryo = kryoThreadLocal.get();
            kryo.writeObject(output, object);
            kryoThreadLocal.remove();
            return output.toBytes();
        } catch (IOException e) {
            log.error("kryo序列化出现异常：{}", e.getMessage());
            throw new RpcException(RpcError.SERIALIZE_FAILED);
      }
    }

    @Override
    public Object deserialize(byte[] bytes, Class<?> clazz) {
        try (
                ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
                Input input = new Input(bais);
        ) {
            Kryo kryo = kryoThreadLocal.get();
            Object object = kryo.readObject(input, clazz);
            kryoThreadLocal.remove();
            return object;
        } catch (IOException e) {
            log.error("kryo反序列化出现异常：{}", e.getMessage());
            e.printStackTrace();
        }
        return null;
    }

    @Override
    public int getCode() {
        return 0;
    }
}
